package com.bfxy.kafka.api.serial;

import com.bfxy.kafka.api.model.User;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class UserDeserializer implements Deserializer<User> {

    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public User deserialize(String s, byte[] bytes) {
        if (null == bytes) {
            return null;
        }
        if (bytes.length < 8) {
            throw new SerializationException("size is wrong, must be data.length >= 8");
        }
        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        // idBytes字节数组的真实长度
        int idLen = buffer.getInt();
        byte[] idBytes = new byte[idLen];
        buffer.get(idBytes);

        // nameBytes字节数组的真实长度
        int nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);

        String id,name;

        try {
            id = new String(idBytes, StandardCharsets.UTF_8);
            name = new String(nameBytes, StandardCharsets.UTF_8);
        } catch (Exception e) {
            throw new SerializationException("deserializing error!", e);
        }
        return new User(id, name);
    }

    @Override
    public void close() {

    }

}
